package com.zz21

import com.gosun.{DataEncry, getResult}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.storage.StorageLevel

import scala.collection.JavaConversions.asScalaBuffer
import scala.collection.mutable.ArrayBuffer

object zz21_vw_censusregisterfamilys {
 // Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession
      .builder()
      .config("spark.network.timeout", "1200")
      .config("spark.kryoserializer.buffer.max","2048")
      //.master("local[*]")
      .appName("SQLContextApp")
      .getOrCreate()
    val options = Map(
      "es.nodes.wan.only" -> "true",
      "es.nodes" -> "192.168.130.87",
      "es.port" -> "9200",
      "es.read.field.as.array.include" -> "arr1, arr2",
      "es.scroll.size" -> "10000",
      "es.input.use.sliced.partitions" -> "false"
    )
    val index = "zz21_vw_censusregisterfamilys"
    val frame: DataFrame = sparkSession
      .read
      .format("es")
      .options(options)
      .load(index)
      .persist(StorageLevel.MEMORY_AND_DISK)
    //frame.show()

    val resSchema = StructType(
      List(
        StructField("CURRENTADDRESS", StringType, true),
        StructField("TELEPHONE", StringType, true),
        StructField("MOBILENUMBER", StringType, true),
        StructField("UPDATEDATE", StringType, true),
        StructField("CREATEDATE", StringType, true),
        StructField("UPDATEUSER", StringType, true),
        StructField("CREATEUSER", StringType, true),
        StructField("DISTRICT", StringType, true),
        StructField("CITY", StringType, true),
        StructField("PROVINCE", StringType, true),
        StructField("HOUSESTATUS", StringType, true),
        StructField("HOMEPHONE", StringType, true),
        StructField("HOUSEHOLDNAME", StringType, true),
        StructField("IDCARDNO", StringType, true),
        StructField("ACCOUNTNUMBER", StringType, true),
        StructField("ORGID", StringType, true),
        StructField("ID", StringType, true),
        StructField("LOGOUT", StringType, true),
        StructField("ORGINTERNALCODE", StringType, true),
        StructField("TELEPHONE_AES", StringType, true),
        StructField("MOBILENUMBER_AES", StringType, true),
        StructField("HOMEPHONE_AES", StringType, true),
        StructField("IDCARDNO_AES", StringType, true)


      )
    )
    val value: RDD[Row] = frame.rdd.mapPartitions(iter => {
      val list = ArrayBuffer[Row]()
      while (iter.hasNext) {
        val row: Row = iter.next()
        val str1 = DataEncry.changAddress(row.getAs[String]("CURRENTADDRESS"))
        val str2 = DataEncry.changPhone(row.getAs[String]("TELEPHONE"))
        val str3 = DataEncry.changPhone(row.getAs[String]("MOBILENUMBER"))
        val str4 = row.getAs[String]("UPDATEDATE")
        val str5 = row.getAs[String]("CREATEDATE")
        val str6 = DataEncry.changName(row.getAs[String]("UPDATEUSER"))
        val str7 = DataEncry.changName(row.getAs[String]("CREATEUSER"))
        val str8 = row.getAs[String]("DISTRICT")
        val str9 = row.getAs[String]("CITY")
        val str10 = row.getAs[String]("PROVINCE")
        val str11 = row.getAs[String]("HOUSESTATUS")
        val str12 = DataEncry.changPhone(row.getAs[String]("HOMEPHONE"))
        val str13 = DataEncry.changName(row.getAs[String]("HOUSEHOLDNAME"))
        val str14 = DataEncry.changIDcard(row.getAs[String]("IDCARDNO"))
        val str15 = row.getAs[String]("ACCOUNTNUMBER")
        val str16 = row.getAs[String]("ORGID")
        val str17 = row.getAs[String]("ID")
        val str18 = row.getAs[String]("LOGOUT")
        val str19 = row.getAs[String]("ORGINTERNALCODE")
        val str20 = DataEncry.changAES(row.getAs[String]("TELEPHONE"))
        val str21 = DataEncry.changAES(row.getAs[String]("MOBILENUMBER"))
        val str22 = DataEncry.changAES(row.getAs[String]("HOMEPHONE"))
        val str23 = DataEncry.changAES(row.getAs[String]("IDCARDNO"))
        val schema: GenericRowWithSchema = new GenericRowWithSchema(Array(
          str1,
          str2,
          str3,
          str4,
          str5,
          str6,
          str7,
          str8,
          str9,
          str10,
          str11,
          str12,
          str13,
          str14,
          str15,
          str16,
          str17,
          str18,
          str19,
          str20,
          str21,
          str22,
          str23
        ), resSchema)
        list.append(schema)
      }
      list.iterator
    })
    val dataFrame: DataFrame = sparkSession.createDataFrame(value.coalesce(1).cache(), resSchema)
    val path = "/kkk/yichun/zz21_vw_censusregisterfamilys/"
    val indespath = path + index + ".json"
    dataFrame.show(false)
    val str: String = dataFrame.toJSON.collectAsList.mkString("[", ",", "]")
    //执行方法
    println("执行方法")
    getResult.getData(str, path)

    println("输出完成！")
    sparkSession.close()

  }

}
